package com.song.sparkstudy.sql;

import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.examples.ml.DataFrameExample;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import com.song.sparkstudy.stream.bean.Person;

public class JavaSparkSQL {
	
	
	public static void main(String[] args) {
		SparkConf conf = new SparkConf().setAppName("app sql").setMaster("local[2]");
		JavaSparkContext ctx =  new JavaSparkContext(conf);
		
		SQLContext sqlcontext = new SQLContext(ctx);
		
//		testTextDataset(ctx , sqlcontext);
		testJsonDataset(ctx ,sqlcontext);
//		testHiveDataset(ctx);
//		testJdbcDataset(ctx ,sqlcontext);
		
	}

	private static void testHiveDataset(JavaSparkContext ctx) {
		// TODO Auto-generated method stub
		
	}

	private static void testJdbcDataset(JavaSparkContext ctx, SQLContext sqlcontext) {
		// TODO Auto-generated method stub
		
	}

	private static void testJsonDataset(JavaSparkContext ctx, SQLContext sqlcontext) {
		
		DataFrame personjson  = sqlcontext.read().json("/tmp/people.json"); 
		personjson.printSchema();
		
		personjson.registerTempTable("people");
		DataFrame testf = sqlcontext.sql("select name ,age from people where age >10 and age <30");
		
		dataframe2rddandprint(testf);
	}

	private static void testTextDataset(JavaSparkContext ctx, SQLContext sqlcontext) {
		System.out.println("=======  data source rdd ======");
		
		JavaRDD<Person> persons =  ctx.textFile("/tmp/people").map(new Function<String, Person>(){

			/**
			 * 
			 */
			private static final long serialVersionUID = 1L;

			@Override
			public Person call(String line) throws Exception {
				
				String[] parts =  line.split(",");
				Person p = new Person();
				p.setName(parts[0]);
				p.setAge(Integer.parseInt(parts[1].trim()));
				return p;
			}
		});
		
		DataFrame pdataframe  = sqlcontext.createDataFrame(persons, Person.class);
		pdataframe.registerTempTable("person");
		
		DataFrame qdataframe = sqlcontext.sql("select name ,age  from person where age >10 and age <20");
		
		dataframe2rddandprint(qdataframe);
	}

	private static void dataframe2rddandprint(DataFrame qdataframe) {
		
		List<String> res = qdataframe.toJavaRDD().map(new Function<Row, String>() {

			@Override
			public String call(Row row) throws Exception {
				return "name is :"+row.getString(0)+ " and age is "+row.getLong(1);
			}
			
		}).collect();
		
		
		for(String val :res)
		{
			System.out.println("  data frame is  "+ val );
		}
		
	}
	
	
}
